package com.dxf.bigdata.D05_spark_again.存储

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * checkpoint 检查点 ,指定落盘路径(persist 虽然也落盘,但是是临时文件会被删除,checkpoint不会)
 *
 * checkpoint 会改变血缘关系
 * cache 不会
 */
object checkpoint2 {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("app")


    sparkConf.set("spark.port.maxRetries", "100")

    val sc = new SparkContext(sparkConf)
    sc.setCheckpointDir("checkpointDir")

    //word count
    val rdd: RDD[String] = sc.makeRDD(List("hello spark", "hello word"))

    val splitRDD: RDD[String] = rdd.flatMap(line => {
      line.split(" ")
    })

    val mapRDD: RDD[(String, Int)] = splitRDD.map(x => {
      (x, 1)
    })


    //    mapRDD.cache()

    mapRDD.checkpoint()
    println(mapRDD.toDebugString)

    println("#########################################################")


    val value: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    //行动算子触发cache缓存,会修改血缘关系
    // 触发checkpoint 修改数据源
    value.collect().foreach(println)

    println(mapRDD.toDebugString)

  }

}
